package net.wzw.mq.consumer.listener;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.extern.slf4j.Slf4j;
import net.wzw.constant.RedisKeyConstant;
import net.wzw.entity.FoodListEntity;
import net.wzw.entity.FoodStepDetailEntity;
import net.wzw.enums.FoodCateTag;
import net.wzw.enums.FoodStatus;
import net.wzw.enums.FoodTime;
import net.wzw.mapper.FoodMapper;
import net.wzw.model.FoodDO;
import net.wzw.mq.FoodUpdateMQMessage;
import net.wzw.redis.RedisCache;
import net.wzw.redis.RedisLock;
import net.wzw.utils.JsonUtil;
import net.wzw.vo.FoodVO;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @author wzw
 * @date 2025/4/21
 * @Description 消费者监听器并发消费
 */
@Component
@Slf4j
public class FoodUpdateListener implements MessageListenerConcurrently {

    /**
     *
     * @param list
     * @param consumeConcurrentlyContext
     * @return 每页 20 条
     */
    private static final int PAGE_SIZE = 20;

    /**
     * 注入 redis
     * **/
    @Autowired
    private RedisCache redisCache;

    @Autowired
    private RedisLock redisLock;

    @Autowired
    private FoodMapper foodMapper;

    /***
     * 并发消费美食消息
     * **/
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

        try {
            for (MessageExt messageExt : msgList) {

                String msg = new String(messageExt.getBody());
                //解析美食信息
                FoodUpdateMQMessage message = JsonUtil.json2Object(msg, FoodUpdateMQMessage.class);
                log.info("美食MQ异步更新-执行美食博主美食分页缓存数据异步更新,消息内容:{}", message);
                //美食博主id
                Long userId = message.getUserId();
                //这里增加分布式锁。同一个用户同时间只能操作一次。避免重复请求
                String foodUpdateLockKey = RedisKeyConstant.FOOD_UPDATE_LOCK_PREFIX + userId;
                //这里通过阻塞的方式进行加锁,跟新增/更新美食操作互斥，并发操作时进行阻塞
                redisLock.blockedLock(foodUpdateLockKey);
                //TODO 并发消费消息逻辑
                try {
                    //此时我们需要对美食闭住的美食列表进行分页缓存重新构建
                    // 1.计算我的美食列表总数
                    String userFoodTotalKey = RedisKeyConstant.USER_FOOD_TOTAL_PRXFIX + userId;

                    int size = redisCache.getInt(userFoodTotalKey);

                    log.info("美食MQ异步更新key:{},value:{}", userFoodTotalKey, size);

                    //计算总分页数
                    int pageNums = (int) Math.ceil((double) size / PAGE_SIZE) + 1;
                    //2.循环构建美食分页缓存
                    for (int page = 1; page <= pageNums; page++) {
                        //美食分页缓存 key
                        String userFoodPageKey = RedisKeyConstant.USER_FOOD_PAGE_PREFIX + userId + ":" + page;
                        //先从缓存中获取下
                        String foodCache = redisCache.get(userFoodPageKey);
                        // 如果分页缓存为空，跳过
                        if (foodCache == null || "".equals(foodCache)) {

                            continue;
                        }
                        //只有缓存中存在这一页的数据才回去数据库读取最新的美食信息去更新
                        //分页信息
                        Page<FoodDO> pageInfo = new Page<>(page, pageNums);

                        IPage<FoodDO> foodDOIPage = null;

                        //查询我的未删除美食列表
                        foodDOIPage = foodMapper.selectPage(pageInfo, new QueryWrapper<FoodDO>().eq("user_id", userId).eq("food_status", FoodStatus.DEFAULT_STATUS));

                        List<FoodDO> foodDOList = foodDOIPage.getRecords();

                        log.info("美食MQ异步更新-从数据库获取我的美食列表,page:{},total:{},size:{}", foodDOIPage.getPages(), foodDOIPage.getTotal(), foodDOIPage.getSize());

                        //对象流式转换处理美食列表结果信息
                        List<FoodVO> foodVOList = foodDOList.stream().map(obj -> {
                            FoodVO foodVO = new FoodVO();
                            if (obj.getFoodCateTag() != null) {
                                //设置美食分类
                                foodVO.setFoodCateTag(FoodCateTag.getByTag(obj.getFoodCateTag()).getTagValue());

                            }
                            if (obj.getFoodDifficulty() != null) {

                                //设置美食制作时常
                                foodVO.setFoodTime(FoodTime.getByTime(obj.getFoodTime()).getTimeValue());

                            }
                            if (StringUtils.isNotBlank(obj.getFoodDetail())) {

                                //设置美食详情
                                List<FoodStepDetailEntity> stepDetails = JsonUtil.StringtoList(obj.getFoodDetail(), FoodStepDetailEntity.class);
                                foodVO.setFoodStepDetailEntityList(stepDetails);

                            }
                            if (StringUtils.isNotBlank(obj.getFoodList())) {
                                // 设置美食清单
                                List<FoodListEntity> foodList = JsonUtil.StringtoList(obj.getFoodList(), FoodListEntity.class);
                                foodVO.setFoodList(foodList);
                            }
                            // TODO 商品信息关联
                            if (StringUtils.isNotBlank(obj.getFoodSkuids())) {
                                List<Long> skuIds = JsonUtil.jsonToLongList(obj.getFoodSkuids());
                                // 设置美食关联商品id
                                foodVO.setSkuIds(skuIds);
                            }
                            BeanUtils.copyProperties(obj, foodVO);
                            return foodVO;
                        }).collect(Collectors.toList());

                        //写入分页缓存数据
                        redisCache.set(userFoodPageKey, JsonUtil.object2Json(foodVOList), RedisCache.generateCacheExpire());
                    }


                } finally {
                    redisLock.unlock(foodUpdateLockKey);
                }

            }
        }catch (Exception e){
            //本次消费失败。下次重新消费
            log.error("美食MQ异步更新 consume error,更新博主美食缓存数据消费失败", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        log.info("美食MQ异步更新-博主美食缓存数据消费成功,result:{}", ConsumeConcurrentlyStatus.CONSUME_SUCCESS);

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

    }

}
